package com.ics.top.batch.management.config;


import com.ics.top.batch.management.domain.BatchUserManagementDTO;
import com.ics.top.batch.management.listener.JobUserManagementListener;
import com.ics.top.batch.management.listener.UserManagementLoggerListener;
import com.ics.top.batch.management.processor.BatchUserManagementProcessor;
import com.ics.top.batch.management.rowmapper.BatchUserManagementRowMapper;
import com.ics.top.config.datasource.DataSourceConfig;
import com.ics.top.config.staticvariable.StaticVariable;
import com.ics.top.second.domain.UserManagement;
import com.ics.top.service.util.DateUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.JdbcCursorItemReader;
import org.springframework.batch.item.database.JpaItemWriter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.orm.jpa.JpaTransactionManager;
import org.springframework.transaction.PlatformTransactionManager;

import javax.persistence.EntityManager;
import javax.persistence.EntityManagerFactory;
import javax.persistence.PersistenceContext;
import javax.sql.DataSource;

/**
 * Created by Administrator on 2018/4/25.
 */

@Configuration
@EnableBatchProcessing
@ComponentScan(basePackageClasses = DataSourceConfig.class)
public class BatchUserManagementConfiguration {
    private static final Logger log = LoggerFactory.getLogger(BatchUserManagementConfiguration.class);
    @Autowired
    public JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;
    @Autowired
    @Qualifier("entityManagerSecondary")
    private EntityManager em;

    @Bean
    public ItemReader<BatchUserManagementDTO> readerUserManagement( @Qualifier("primaryDataSource") DataSource dataSource,BatchUserManagementRowMapper userRowMapper)  {
        JdbcCursorItemReader itemReader = new JdbcCursorItemReader();
        itemReader.setDataSource(dataSource);
        String time  = DateUtil.getDatePeriod(1);
        String sql = " SELECT " +
            " a.id, a.avg_price, a.biz_order_id,a.buyer_nick, " +
            " a.city,a.close_trade_amount,a.close_trade_count, " +
            " a.grade,a.grade_name,a.item_close_count, a.item_num, " +
            " a.last_trade_time,  a.province, a.relation_source,  a.`status`, " +
            " a.trade_amount, a.trade_count, a.stores_name,a.group_ids,a.locale,a.mobile " +
            " FROM " +
            " sys_member a " +
            " LEFT JOIN batch_user_management_logs b ON b.management_ids LIKE CONCAT('%,', a.id, ',%') " +
            " WHERE " +
            " ISNULL(b.id) ";
        log.info("查询sql语句"+sql);
        itemReader.setSql(sql);
        itemReader.setRowMapper(userRowMapper);
        itemReader.close();
        return itemReader;
    }

    @Bean
    public BatchUserManagementProcessor processorUserManagement(BatchUserManagementProcessor userItemProcessor) {
        return  userItemProcessor;
    }

    @Bean
    public ItemWriter<UserManagement> writerUserManagement() {
        JpaItemWriter writer = new JpaItemWriter<UserManagement>();
        try {
            writer.setEntityManagerFactory(em.getEntityManagerFactory());
            writer.afterPropertiesSet();
        } catch (Exception e) {
            e.printStackTrace();
        }

        return writer;
    }

    @Bean
    public Job userManagementJob1(JobUserManagementListener listener, Step userManagementStep1 ) {
        return jobBuilderFactory.get("userManagementJob1")//获取job
            .incrementer(new RunIdIncrementer())//
            .listener(listener)
            .start(userManagementStep1)
            .build();
    }
    @Bean
    public PlatformTransactionManager userManagementTransactionManager() {
        return new JpaTransactionManager(em.getEntityManagerFactory());
    }

    @Bean
    public Step userManagementStep1(
        ItemReader<BatchUserManagementDTO> readerUserManagement,
        ItemWriter<UserManagement> writerUserManagement,
        BatchUserManagementProcessor processorUserManagement,
        UserManagementLoggerListener itemLoggerListener,
        @Qualifier("userManagementTransactionManager")PlatformTransactionManager userManagementTransactionManager ) {
        return stepBuilderFactory.get("userManagementStep1")
            .transactionManager(userManagementTransactionManager)
            .listener(itemLoggerListener)
            .<BatchUserManagementDTO, UserManagement> chunk(StaticVariable.BATCH_CHUNK_COUNT)
            .reader(readerUserManagement)
            .processor(processorUserManagement)//处理数据
            .writer(writerUserManagement) //写入数据
            .build();
    }
}
